package drds.global_transaction.transaction_coordinator.store;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.TimeUnit;

import drds.common.$;
import drds.global_transaction.api.api.GlobalTransactionStatus;
import drds.global_transaction.transaction_coordinator.transaction_session.BranchTransactionSession;
import drds.global_transaction.transaction_coordinator.transaction_session.GlobalTransactionSession;
import drds.global_transaction.transaction_coordinator.transaction_session.TransactionSessionCondition;
import lombok.extern.slf4j.Slf4j;

@Slf4j
class WriteDataFileRunnable implements Runnable
{
	private FileTransactionStoreManager fileTransactionStoreManager;

	public WriteDataFileRunnable(FileTransactionStoreManager fileTransactionStoreManager)
	{
		this.fileTransactionStoreManager = fileTransactionStoreManager;
	}

	@Override
	public void run()
	{
		while (!fileTransactionStoreManager.stop)
		{
			try
			{
				TransactionWriteFuture transactionWriteFuture = fileTransactionStoreManager.transactionWriteFutureQueue.poll(FileTransactionStoreManager.max_pool_time_mills, TimeUnit.MILLISECONDS);
				if (null == transactionWriteFuture)
				{
					flushWhenSatisfyCondition();
					continue;
				}
				if (transactionWriteFuture.isTimeout())
				{
					transactionWriteFuture.setResult(Boolean.FALSE);
					continue;
				}
				if (writeData(transactionWriteFuture.getTransactionWriteStore().encode()))
				{
					transactionWriteFuture.setResult(Boolean.TRUE);
					FileTransactionStoreManager.transaction_sequence_number.incrementAndGet();
					flushWhenSatisfyCondition();
				} else
				{
					transactionWriteFuture.setResult(Boolean.FALSE);
				}
				if (FileTransactionStoreManager.transaction_sequence_number.get() % FileTransactionStoreManager.per_file_block_size == 0 && (System.currentTimeMillis() - FileTransactionStoreManager.startTimeMills) > FileTransactionStoreManager.max_transaction_timeout_mills)
				{
					saveHistory();
				}
			} catch (InterruptedException exx)
			{
				fileTransactionStoreManager.stop = true;
			} catch (Exception exx)
			{
				log.error(exx.getMessage());
			}
		}

	}

	private void saveHistory() throws IOException
	{
		try
		{
			List<GlobalTransactionSession> globalTransactionSessionList = fileTransactionStoreManager.transactionSessionManager.getGlobalTransactionSessionList(new TransactionSessionCondition(GlobalTransactionStatus.begin, FileTransactionStoreManager.max_transaction_timeout_mills));
			if ($.isHasData(globalTransactionSessionList))
			{
				for (GlobalTransactionSession globalTransactionSession : globalTransactionSessionList)
				{
					TransactionWriteStore globalTransactionSessionTransactionWriteStore = new TransactionWriteStore(globalTransactionSession, OperationType.add_global_transaction);
					writeData(globalTransactionSessionTransactionWriteStore.encode());
					List<BranchTransactionSession> branchTransactionSessionList = globalTransactionSession.getSortedBranchTransactionSessionList();
					if (null != branchTransactionSessionList)
					{
						for (BranchTransactionSession branchTransactionSession : branchTransactionSessionList)
						{
							TransactionWriteStore branchTransactionSessionTransactionWriteStore = new TransactionWriteStore(branchTransactionSession, OperationType.add_branch_transaction);
							writeData(branchTransactionSessionTransactionWriteStore.encode());
						}
					}
				}
			}
			fileTransactionStoreManager.currentFileChannel.force(true);
			File historyFile = new File(fileTransactionStoreManager.historyFullFileName);
			if (historyFile.exists())
			{
				historyFile.delete();
			}
			fileTransactionStoreManager.closeFile(fileTransactionStoreManager.currentFileChannel, fileTransactionStoreManager.currentRandomAccessFile);
			fileTransactionStoreManager.currentFile.renameTo(new File(fileTransactionStoreManager.historyFullFileName));
		} catch (IOException exx)
		{
			log.error("save history data file error," + exx.getMessage());
		} finally
		{
			fileTransactionStoreManager.initFile(fileTransactionStoreManager.currentFileFullName);
		}

	}

	private void flushWhenSatisfyCondition()
	{
		if (!FileTransactionStoreManager.enable_schedule_flush)
		{
			return;
		}
		long interval = FileTransactionStoreManager.transaction_sequence_number.get() - FileTransactionStoreManager.file_flush_transaction_sequence_number.get();
		if (interval == 0)
		{
			return;
		}
		if (interval % FileTransactionStoreManager.max_flush_num == 0 || System.currentTimeMillis() - fileTransactionStoreManager.currentFile.lastModified() > FileTransactionStoreManager.max_flush_time_mills)
		{
			try
			{
				fileTransactionStoreManager.currentFileChannel.force(false);
			} catch (IOException exx)
			{
				log.error("flush error:" + exx.getMessage());
			}
			FileTransactionStoreManager.file_flush_transaction_sequence_number.addAndGet(interval);
		}
	}

	private boolean writeData(byte[] bytes)
	{
		int retry = 0;
		byte[] realWriteBytes = new byte[bytes.length + 4];
		ByteBuffer byteBuffer = ByteBuffer.wrap(realWriteBytes);
		byteBuffer.putInt(bytes.length);
		byteBuffer.put(bytes);
		for (; retry < FileTransactionStoreManager.max_write_retry; retry++)
		{
			try
			{
				//
				byteBuffer.flip();
				while (byteBuffer.hasRemaining())
				{
					fileTransactionStoreManager.currentFileChannel.write(byteBuffer);
				}
				return true;
			} catch (IOException exx)
			{
				log.error("write data file error:" + exx.getMessage());
			}
		}
		log.error("write dataFile failed,retry more than :" + FileTransactionStoreManager.max_write_retry);
		return false;
	}

}
